package com.sunday.common.reactor.e3_Creating_New_Sequence;

import com.sunday.common.reactor.SubscribeUtils;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.LocalDateTime;
import java.util.Arrays;
import java.util.Optional;

/**
 * A.1. 创建一个新序列...
 */
@Slf4j
public class CreatingNewSequence {

    /**
     * 发出一个 ，我已经有：（Flux|单T just)
     * Optional<T>： Mono#justOrEmpty（Optional<T>）
     * T：Mono#justOrEmpty（T）null
     */
    @Test
    public void monoJust() {
        log.info("Mono.just 会抛出异常 java.lang.NullPointerException: value 不能为null 。");
        // Mono.just 会抛出异常 java.lang.NullPointerException: value 不能为null 。
        Mono<String> mono = Mono.just("foo");
        SubscribeUtils.subscribe(mono);
        log.info("Mono.justOrEmpty 会忽略 null 值，而不是抛出异常。");
        // Mono.justOrEmpty 会忽略 null 值，而不是抛出异常。
        mono = Mono.justOrEmpty(Optional.empty());
        SubscribeUtils.subscribe(mono);
        mono = Mono.justOrEmpty(null);
        SubscribeUtils.subscribe(mono);
        log.info("java.lang.NullPointerException: value");
        // java.lang.NullPointerException: value
        mono = Mono.just(null);
        SubscribeUtils.subscribe(mono);
    }

    /**
     * 发出由方法返回的：（Flux|单声道）以及T just
     * 但懒惰地捕获：使用 Mono#fromSupplier 或 wrap （Flux|Mono） 内部 （Flux|单just defer)
     */
    @Test
    public void fluxJust() {

        log.info("Mono.fromSupplier 用于创建 Mono 对象。它接受一个 Supplier（供应者）函数作为参数，该函数在每次订阅时都会被调用，以生成一个值。这样可以实现对值的延迟提供，直到订阅时才进行实际的值生成。");
        // Mono.fromSupplier 会在订阅时调用提供的方法。
        // Mono.fromSupplier 是 Reactor 框架中的一个功能，用于创建 Mono 对象。它接受一个 Supplier（供应者）函数作为参数，该函数在每次订阅时都会被调用，以生成一个值。这样可以实现对值的延迟提供，直到订阅时才进行实际的值生成。
        Mono<String> mono = Mono.fromSupplier(() -> LocalDateTime.now().toString());
        SubscribeUtils.subscribe(mono);
        SubscribeUtils.subscribe(mono);
        log.info("Flux.just 会抛出异常 java.lang.NullPointerException: value 不能为null 。");
        // Flux.just 会抛出异常 java.lang.NullPointerException: value 不能为null 。
        Flux<String> flux = Flux.just("foo", "bar", "foobar");
        SubscribeUtils.subscribe(flux);
        log.info("Flux.defer 它允许您定义一个函数，该函数在每次订阅时都会被调用，以生成一个新的 Flux 对象。这样可以实现对数据流的延迟创建，直到订阅时才进行实际的数据处理和事件触发。");
        // Flux.defer 会在每次订阅时调用提供的方法。
        // 当使用 Flux.defer 时，它允许您定义一个函数，该函数在每次订阅时都会被调用，以生成一个新的 Flux 对象。这样可以实现对数据流的延迟创建，直到订阅时才进行实际的数据处理和事件触发。
        flux = Flux.defer(() -> Mono.just(LocalDateTime.now().toString()));
        SubscribeUtils.subscribe(flux);
        SubscribeUtils.subscribe(flux);

    }

    /**
     * 发出几个我可以显式枚举的：Flux#just（T...）T
     * 迭代：
     * 数组：Flux#fromArray
     * 集合或可迭代：Flux#fromIterable
     * 整数范围：Flux#range
     * 为每个订阅提供的流：Flux#fromStream（Supplier<Stream>）
     */
    @Test
    public void fluxJustArr() {
        log.info("Flux.just 会抛出异常 java.lang.NullPointerException: value 不能为null 。");
        // Flux.just 会抛出异常 java.lang.NullPointerException: value 不能为null 。
        Flux<String> flux = Flux.just("foo", "bar", "foobar");
        SubscribeUtils.subscribe(flux);
        log.info("Flux.fromArray 会抛出异常 java.lang.NullPointerException: value 不能为null 。");
        // Flux.fromArray 会抛出异常 java.lang.NullPointerException: value 不能为null 。
        String[] array = new String[]{"foo", "bar", "foobar"};
        flux = Flux.fromArray(array);
        SubscribeUtils.subscribe(flux);
        log.info("Flux.fromIterable 会抛出异常 java.lang.NullPointerException: value 不能为null 。");
        // Flux.fromIterable 会抛出异常 java.lang.NullPointerException: value 不能为null 。
        flux = Flux.fromIterable(Arrays.asList(array));
        SubscribeUtils.subscribe(flux);
        log.info("Flux.range 会抛出异常 java.lang.NullPointerException: value 不能为null 。");
        // Flux.range 会抛出异常 java.lang.NullPointerException: value 不能为null 。
        flux = Flux.range(1, 5).map(i -> "foo - " + i);
        SubscribeUtils.subscribe(flux);
        log.info("Flux.fromStream 会抛出异常 java.lang.NullPointerException: value 不能为null 。");
        // Flux.fromStream 会抛出异常 java.lang.NullPointerException: value 不能为null 。
        flux = Flux.fromStream(Arrays.asList(array).stream());
        SubscribeUtils.subscribe(flux);
    }

    /**
     * 从各种单值源发出，例如：
     * a 供应商<T>： Mono#fromSupplier
     * 任务：Mono#fromCallable、Mono#fromRunnable
     * a CompletableFuture<T>： 单声道#fromFuture
     */
    @Test
    public void monoFrom() {
        log.info("Mono.fromSupplier 会在订阅时调用提供的方法。");
        // Mono.fromSupplier 会在订阅时调用提供的方法。
        Mono<String> mono = Mono.fromSupplier(() -> LocalDateTime.now().toString());
        SubscribeUtils.subscribe(mono);
        SubscribeUtils.subscribe(mono);
        log.info("Mono.fromCallable 会在订阅时调用提供的方法。");
        // Mono.fromCallable 会在订阅时调用提供的方法。
        mono = Mono.fromCallable(() -> LocalDateTime.now().toString());
        SubscribeUtils.subscribe(mono);
        SubscribeUtils.subscribe(mono);
        log.info("Mono.fromRunnable 会在订阅时调用提供的方法。");
        // Mono.fromRunnable 会在订阅时调用提供的方法。
        mono = Mono.fromRunnable(() -> log.info("[Mono.fromRunnable] : {}", LocalDateTime.now().toString()));
        SubscribeUtils.subscribe(mono);
        SubscribeUtils.subscribe(mono);
        log.info("Mono.fromFuture 会在订阅时调用提供的方法。");
        // Mono.fromFuture 会在订阅时调用提供的方法。
        mono = Mono.fromFuture(() -> Mono.just(LocalDateTime.now().toString()).toFuture());
        SubscribeUtils.subscribe(mono);
        SubscribeUtils.subscribe(mono);
    }

    /**
     * 完成：（Flux|单empty)
     * 立即出错：（Flux|单error)
     * 但懒惰地构建 Throwable：（Flux|单error(Supplier<Throwable>))
     * 从不做任何事情：（Flux|单never)
     * 这是在订阅时决定的：（Flux|单defer)
     * 这取决于一次性资源：（Flux|单using)
     * 以编程方式生成事件（可以使用状态）：
     * 同步和逐个：Flux#generate
     * 异步（也可以同步），一次可以进行多次发射：Flux#create（Mono#create，没有多重发射方面）
     */
    @Test
    public void fluxFrom() {
        log.info("Flux.empty 会发出完成信号，而不发出任何元素。");
        // Flux.empty 会发出完成信号，而不发出任何元素。
        Flux<String> flux = Flux.empty();
        SubscribeUtils.subscribe(flux);
        log.info("Flux.error 会发出错误信号，而不发出任何元素。");
        // Flux.error 会发出错误信号，而不发出任何元素。
        flux = Flux.error(new RuntimeException("error"));
        SubscribeUtils.subscribe(flux);
        log.info("Flux.error 会发出错误信号，而不发出任何元素。");
        // Flux.error 会发出错误信号，而不发出任何元素。
        flux = Flux.error(() -> new RuntimeException("error"));
        SubscribeUtils.subscribe(flux);
        log.info("Flux.never 不会发出任何信号。");
        // Flux.never 不会发出任何信号。
        flux = Flux.never();
        SubscribeUtils.subscribe(flux);
        log.info("Flux.defer 会在每次订阅时调用提供的方法。");
        // Flux.defer 会在每次订阅时调用提供的方法。
        flux = Flux.defer(() -> Flux.just(LocalDateTime.now().toString()));
        SubscribeUtils.subscribe(flux);
        SubscribeUtils.subscribe(flux);
        log.info("Flux.using 会在订阅时调用提供的方法。 try-with-resource");
        // Flux.using 会在订阅时调用提供的方法。 try-with-resource
        flux = Flux.using(() -> "foo", s -> Flux.just(s), s -> log.info("[Flux.using] : {}", s));
        SubscribeUtils.subscribe(flux);
        log.info("Flux.generate 会在订阅时调用提供的方法。");
        // Flux.generate 会在订阅时调用提供的方法。
        flux = Flux.generate(sink -> {
            sink.next("foo");
            sink.complete();
        });
        SubscribeUtils.subscribe(flux);
        log.info("Flux.create 会在订阅时调用提供的方法。");
        // Flux.create 会在订阅时调用提供的方法。
        flux = Flux.create(sink -> {
            sink.next("foo");
            sink.complete();
        });
        SubscribeUtils.subscribe(flux);
    }

}
